Add: generation-safe two-frame local endpoint - #1574
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughPipeline leases and run-scoped queues implement bounded whole-run FIFO admission. Framed mailboxes support depth-aware concurrent dispatch, while bindings, documentation, and tests expose and validate the new protocol. ChangesPipeline admission and dispatch
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/common/hierarchical/scheduler.cpp (1)
465-492: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
breakafter one dispatch keeps the second task frame idle.
has_capacity()now allows two in-flight tasks per NEXT_LEVEL worker, but the loop still dispatches at most one slot per worker perdispatch_ready()pass. If both tasks of a run are already READY and enqueued, the second frame is only filled on the next wake (a completion or a freshnotify_ready), so the intended overlap is lost for graphs that enqueue both dispatches before the scheduler runs.♻️ Proposed fix: keep dispatching while capacity remains
s.state.store(TaskState::RUNNING, std::memory_order_release); worker->dispatch(WorkerDispatch{slot, 0}); - break; + if (!worker->has_capacity()) break; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/scheduler.cpp` around lines 465 - 492, Update Scheduler::dispatch_next_level_singles so it continues popping and dispatching READY slots for each worker while worker->has_capacity() remains true, instead of breaking after the first successful dispatch. Preserve the existing validation, run filtering, and misrouting behavior, while ensuring the loop stops when capacity is full or no eligible slot is available.
🧹 Nitpick comments (6)
tests/ut/py/test_worker/test_host_worker.py (1)
256-268: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdd a small sleep to the mailbox spin waits.
Both deadline loops poll with no sleep, so the test thread holds the GIL almost continuously while
_run_chip_main_loop— a Python thread in the same interpreter — needs it to advance the frame state. On a loaded runner this is a real starvation source against the 5s deadline, and other waits in this file already usetime.sleep(0.001).♻️ Proposed change
_mailbox_store_i32(frame1_addr + _OFF_STATE, worker_mod._TASK_READY) deadline = time.monotonic() + 5.0 while _mailbox_load_i32(frame1_addr + _OFF_STATE) != worker_mod._TASK_ACCEPTED_STATE: assert time.monotonic() < deadline + time.sleep(0.001) assert calls == 1 release[0].set() assert entered[1].wait(5.0) assert _mailbox_load_i32(frame0_addr + _OFF_STATE) == worker_mod._TASK_DONE assert _mailbox_load_i32(frame1_addr + _OFF_STATE) == worker_mod._TASK_ACTIVE release[1].set() deadline = time.monotonic() + 5.0 while _mailbox_load_i32(frame1_addr + _OFF_STATE) != worker_mod._TASK_DONE: assert time.monotonic() < deadline + time.sleep(0.001)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/py/test_worker/test_host_worker.py` around lines 256 - 268, Add time.sleep(0.001) inside both deadline polling loops in the mailbox test around _OFF_STATE checks, allowing _run_chip_main_loop to acquire the GIL while preserving the existing 5-second timeout and state assertions.tests/ut/cpp/hierarchical/test_orchestrator.cpp (1)
752-762: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winA regression here hangs the suite instead of failing it.
std::async(std::launch::async, ...)returns a future whose destructor blocks until the task completes. If admission stops being released (the exact regression this test guards),begin_run()never returns,ASSERT_EQat Line 761 aborts the test body, and~futurethen blocks forever — the run that would free the lease is never completed. Consider completing/aborting the blocking work in a scope guard (or driving the release before any fatal assertion) so a failure surfaces as a failed test rather than a stuck CI job. The same shape applies to thereplacementfuture at Lines 793-797.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp` around lines 752 - 762, Make the asynchronous admission checks in the current test, including the “third” and “replacement” futures, failure-safe so a fatal assertion cannot destroy a still-blocked std::future and hang the suite. Add scoped cleanup or perform the lease-release/abort operation before any fatal assertion that depends on future readiness, ensuring blocked begin_run work is always unblocked while preserving the existing success-path assertions.src/common/hierarchical/types.h (1)
154-154: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse
TaskSlotfortask_slotsinstead of a hard-codedint32_t.
Orchestrator::register_run_slotpushesTaskSlotvalues andcancel_unstarted_runcopies them back intostd::vector<TaskSlot>; spelling the element type as the alias keeps the two in step ifTaskSlot's underlying type ever changes.♻️ Proposed change
- std::vector<int32_t> task_slots; + std::vector<TaskSlot> task_slots;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/types.h` at line 154, Update the task_slots member declaration to use the TaskSlot alias as its vector element type instead of hard-coded int32_t, preserving the existing container and behavior.src/common/hierarchical/worker_manager.h (1)
143-149: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a static assert guarding
MAILBOX_ARGS_CAPACITYagainst underflow.The capacity is an unsigned subtraction of two derived offsets; if the header region ever grows past
MAILBOX_OFF_FRAME_PROTOCOLthe result silently wraps to a huge value instead of failing the build. The neighbouring layout invariants already have asserts — this one is worth the same treatment.🛡️ Proposed addition
static constexpr size_t MAILBOX_ARGS_CAPACITY = static_cast<size_t>(MAILBOX_OFF_FRAME_PROTOCOL) - static_cast<size_t>(MAILBOX_OFF_TASK_ARGS_BLOB); +static_assert( + MAILBOX_OFF_TASK_ARGS_BLOB < MAILBOX_OFF_FRAME_PROTOCOL, "task args blob region overlaps the frame identity region" +);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/worker_manager.h` around lines 143 - 149, Add a compile-time assertion immediately after the MAILBOX_ARGS_CAPACITY definition to verify that MAILBOX_OFF_FRAME_PROTOCOL is at least MAILBOX_OFF_TASK_ARGS_BLOB before the unsigned subtraction can underflow. Keep the existing capacity calculation unchanged and follow the neighboring mailbox layout invariant assertion style.python/simpler/worker.py (1)
1912-1912: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReusing the lease
slot_idfield as the depth-publication channel deserves a comment.Startup packs
cw.pipeline_depthinto the lease's slot field and the parent reads it back at line 4747, while every later write to the same offset means "lease slot id". A one-line note here (and at the reader) would keep the dual meaning discoverable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/simpler/worker.py` at line 1912, Add concise comments at the _PIPELINE_LEASE_FMT.pack_into write and its corresponding parent reader near line 4747 documenting that the lease slot_id field temporarily carries pipeline_depth during startup, while later writes use it as the lease slot ID.src/common/hierarchical/worker_manager.cpp (1)
550-566: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueEarly returns inside the frame lock rely on
endpoint_poisoned_for publish-turn escape.The poisoned check (553) and non-IDLE check (559) return before publishing; the first relies on an already-poisoned flag and the second sets it, so waiters escape. Worth an explicit comment stating that invariant, since it is what keeps the
next_publish_id_sequence from stalling here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/common/hierarchical/worker_manager.cpp` around lines 550 - 566, Add a concise comment in the run_two_frame frame-lock early-return block explaining that these publish-skipping exits rely on endpoint_poisoned_ being set, allowing publish-turn waiters to escape without stalling next_publish_id_. Keep the existing poisoned and non-IDLE checks and control flow unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/task-flow.md`:
- Around line 296-328: Remove or rewrite the dormant-capacity paragraph in
docs/task-flow.md lines 296-328 so it no longer describes single-frame, unleased
production behavior and remains consistent with the active depth-two lease/FIFO
contract; update the local mailbox diagram in docs/task-flow.md lines 430-431 to
represent PipelineSlotLease or its {slot_id, generation} fields.
In `@python/simpler/worker.py`:
- Around line 1748-1792: Add a bounded yield to admission_loop after each
polling sweep, including when no control or task frame is admitted, using a very
short sleep or time.sleep(0). Preserve immediate handling of shutdown, control
requests, and ready frames while preventing the admission thread from
continuously contending for the GIL with task execution.
- Line 1762: Update both zip calls in the worker logic— the
frame_bufs/frame_addrs loop and the _chip_shms/_chip_pids loop—to pass
strict=True, preserving the existing iteration while raising an error if the
paired collections differ in length.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 528-548: Ensure every pre-publish exit in
LocalMailboxEndpoint::run_two_frame consumes or invalidates its dispatch
sequence: update next_publish_id_ and notify publish_cv_ before returning on the
no-free-task-frame path, and handle exceptions caught before the publish block
similarly rather than rethrowing with the sequence stranded. If the sequence
cannot be advanced safely, poison the endpoint so later publish waits do not
block indefinitely.
- Around line 582-600: Update the run_two_frame argument-blob handling to check
blob_bytes against MAILBOX_ARGS_CAPACITY before writing the header or payload.
When oversized, fail the completion with ENDPOINT_FAILURE and return before
publishing the frame; retain the existing serialization only for blobs that fit.
In
`@tests/st/a2a3/host_build_graph/worker_async_endpoint/test_worker_async_endpoint.py`:
- Around line 125-132: Update the cleanup in the worker async endpoint test so
host buffers are freed only after run is confirmed terminal: after
run.wait(20.0), recheck run.done and skip or defer st_worker.free_host_buffer
when the run remains active, mirroring the sibling worker async FIFO test.
Preserve the existing exception suppression and buffer cleanup for completed
runs.
---
Outside diff comments:
In `@src/common/hierarchical/scheduler.cpp`:
- Around line 465-492: Update Scheduler::dispatch_next_level_singles so it
continues popping and dispatching READY slots for each worker while
worker->has_capacity() remains true, instead of breaking after the first
successful dispatch. Preserve the existing validation, run filtering, and
misrouting behavior, while ensuring the loop stops when capacity is full or no
eligible slot is available.
---
Nitpick comments:
In `@python/simpler/worker.py`:
- Line 1912: Add concise comments at the _PIPELINE_LEASE_FMT.pack_into write and
its corresponding parent reader near line 4747 documenting that the lease
slot_id field temporarily carries pipeline_depth during startup, while later
writes use it as the lease slot ID.
In `@src/common/hierarchical/types.h`:
- Line 154: Update the task_slots member declaration to use the TaskSlot alias
as its vector element type instead of hard-coded int32_t, preserving the
existing container and behavior.
In `@src/common/hierarchical/worker_manager.cpp`:
- Around line 550-566: Add a concise comment in the run_two_frame frame-lock
early-return block explaining that these publish-skipping exits rely on
endpoint_poisoned_ being set, allowing publish-turn waiters to escape without
stalling next_publish_id_. Keep the existing poisoned and non-IDLE checks and
control flow unchanged.
In `@src/common/hierarchical/worker_manager.h`:
- Around line 143-149: Add a compile-time assertion immediately after the
MAILBOX_ARGS_CAPACITY definition to verify that MAILBOX_OFF_FRAME_PROTOCOL is at
least MAILBOX_OFF_TASK_ARGS_BLOB before the unsigned subtraction can underflow.
Keep the existing capacity calculation unchanged and follow the neighboring
mailbox layout invariant assertion style.
In `@tests/ut/cpp/hierarchical/test_orchestrator.cpp`:
- Around line 752-762: Make the asynchronous admission checks in the current
test, including the “third” and “replacement” futures, failure-safe so a fatal
assertion cannot destroy a still-blocked std::future and hang the suite. Add
scoped cleanup or perform the lease-release/abort operation before any fatal
assertion that depends on future readiness, ensuring blocked begin_run work is
always unblocked while preserving the existing success-path assertions.
In `@tests/ut/py/test_worker/test_host_worker.py`:
- Around line 256-268: Add time.sleep(0.001) inside both deadline polling loops
in the mailbox test around _OFF_STATE checks, allowing _run_chip_main_loop to
acquire the GIL while preserving the existing 5-second timeout and state
assertions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 86fdd0f3-23b4-46cc-82fc-99e1b1d498fa
📒 Files selected for processing (24)
docs/task-flow.mdpython/bindings/task_interface.cpppython/bindings/worker_bind.hpython/simpler/task_interface.pypython/simpler/worker.pysrc/common/hierarchical/orchestrator.cppsrc/common/hierarchical/orchestrator.hsrc/common/hierarchical/scheduler.cppsrc/common/hierarchical/scheduler.hsrc/common/hierarchical/types.cppsrc/common/hierarchical/types.hsrc/common/hierarchical/worker.cppsrc/common/hierarchical/worker.hsrc/common/hierarchical/worker_manager.cppsrc/common/hierarchical/worker_manager.hsrc/common/worker/pipeline_slot_pool.htests/st/a2a3/host_build_graph/worker_async_endpoint/kernels/orchestration/long_vector_orch.cpptests/st/a2a3/host_build_graph/worker_async_endpoint/test_worker_async_endpoint.pytests/st/a2a3/host_build_graph/worker_async_fifo/test_worker_async_fifo.pytests/ut/cpp/hierarchical/test_orchestrator.cpptests/ut/cpp/hierarchical/test_pipeline_contract.cpptests/ut/cpp/hierarchical/test_scheduler.cpptests/ut/py/test_callable_identity.pytests/ut/py/test_worker/test_host_worker.py
789f990 to
be51bfd
Compare
Crane-Liu
changed the title
|
Addressed the review summary and refreshed the branch on the latest main:
Validation on |
- Separate control traffic from two identified task frames - Accept a queued frame while one sequential child executor is active - Bound parent dispatch capacity and fill every available frame - Retire publish ordering across every pre-publish failure path - Reject oversized argument blobs before publishing mailbox state - Preserve single-frame fallbacks on unsupported local backends - Keep failure-path tests and host-buffer cleanup terminal-safe - Release the GIL around native execution so admission remains live
be51bfd to
d6dfaa9
Compare
1 participant
Summary
Dependency
Depends on #1541 (B2 whole-run FIFO admission). Until #1541 merges, this PR intentionally includes that predecessor commit in its diff.
Why TMR remains single-frame
An initial B3 validation enabled the endpoint for TMR as well. Real-device l3_l2_message_queue and l3_l2_orch_comm_stream then reproduced an AIV UB out-of-bounds failure. B3 now gates the endpoint to host_build_graph. TMR prepared-state and RuntimeEnv compatibility belongs to B5; both TMR regressions pass again through the legacy single-frame path.
Validation
Tested commit: be51bfd
No simulation validation was run, per the agreed workflow.